﻿using System;
using SIEGate.Core.MessageBusses;
using SIEGate.Core;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using System.Collections.Generic;
using SIEGate.Core.Measurements;
using SIEGate.Core.Measurements.Values;
using System.Threading.Tasks;
using System.Threading;
using System.Diagnostics;

namespace SIEGateCoreTest.MessageBusTests
{
    [TestClass]
    public class ChannelMessageBusTests
    {
        ChannelMessageBus messageBus;
        static ConsoleTraceListener tracer = new ConsoleTraceListener();

        [ClassInitialize]
        static public void InitClass(TestContext ctx)
        {

            tracer.Name = "debugging";
            Trace.Listeners.Add(tracer);
            tracer.WriteLine("Now Tracing");

        }
        [ClassCleanup]
        static public void CleanupClass()
        {
            Trace.Listeners.Remove(tracer);
        }
        [TestInitialize]
        public void TestsInitialize()
        {
            this.messageBus = new ChannelMessageBus(2 << 11);
        }
        [TestMethod]
        public void TestGetProxy()
        {
            IMessageBusProxy mbp = messageBus.GetProxy("foo");
            Assert.IsNotNull(mbp);
        }
        [TestMethod]
        public void TestSinglePublish()
        {
            IMessageBusProxy mbp = messageBus.GetProxy("foo");
            HashSet<Guid> m = new HashSet<Guid>() { Guid.NewGuid(), Guid.NewGuid() };
            mbp.Publish(m);
        }

        [TestMethod]
        public void TestPublishThenDoItAgain()
        {
            IMessageBusProxy mbp = messageBus.GetProxy("foo");
            HashSet<Guid> m = new HashSet<Guid>(){Guid.NewGuid() , Guid.NewGuid()};
            mbp.Publish(m);
            mbp.Publish(m);
        }

        [TestMethod]
        public void TestSendNoSubscribe()
        {
            IMessageBusProxy mbp = messageBus.GetProxy("foo");
            Guid g1 = Guid.NewGuid();
            HashSet<Guid> pubs = new HashSet<Guid>() { g1, Guid.NewGuid(), Guid.NewGuid() };
            mbp.Publish(pubs);
            IMeasurement m = new Measurement(g1, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                new StringValue("foo"));
            mbp.Send(m);
        }
        [TestMethod]
        public void TestPubThenSub()
        {
            IMessageBusProxy mbp = messageBus.GetProxy("foo");
            IMessageBusProxy mbp2 = messageBus.GetProxy("bar");
            Guid g1 = Guid.NewGuid();
            HashSet<Guid> pubs = new HashSet<Guid>() { g1, Guid.NewGuid(), Guid.NewGuid() };
            mbp.Publish(pubs);
            mbp2.Subscribe(new HashSet<Guid>() { g1 });
        }
        [TestMethod]
        public void TestPubThenSubThenSend()
        {
            IMessageBusProxy mbp = messageBus.GetProxy("foo");
            IMessageBusProxy mbp2 = messageBus.GetProxy("bar");
            Guid g1 = Guid.NewGuid();
            HashSet<Guid> pubs = new HashSet<Guid>() { g1, Guid.NewGuid(), Guid.NewGuid() };
            mbp.Publish(pubs);
            mbp2.Subscribe(new HashSet<Guid>() { g1 });
            IMeasurement m = new Measurement(g1, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
               new StringValue("foo"));
            mbp.Send(m);
            IMeasurement[] recv = mbp2.Recv();
            Assert.AreEqual(recv.Length, 1);
            IMeasurement r = recv[0];
            Assert.ReferenceEquals(m, r);

        }
        [TestMethod]
        public void TestPubThenSubThenSendMany()
        {
            IMessageBusProxy mbp = messageBus.GetProxy("foo");
            IMessageBusProxy mbp2 = messageBus.GetProxy("bar");
            Guid g1 = Guid.NewGuid();
            Guid g2 = Guid.NewGuid();
            HashSet<Guid> pubs = new HashSet<Guid>() { g1, g2, Guid.NewGuid() };
            mbp.Publish(pubs);
            mbp2.Subscribe(new HashSet<Guid>() { g1, g2 });
            IMeasurement[] m = {new Measurement(g1, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
               new StringValue("foo")), new Measurement(g2, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                   new StringValue("bar"))};
            mbp.Send(m);
            IMeasurement[] recv = mbp2.Recv();
            Assert.AreEqual(recv.Length, 2);
            IMeasurement r = recv[0];
            Assert.IsTrue((new HashSet<Guid>() { g1, g2 }).Contains(r.StreamID));

        }
        [TestMethod]
        public void SubBeforePubWithProducerExisting()
        {
            IMessageBusProxy mbp1 = messageBus.GetProxy("one");
            IMessageBusProxy mbp2 = messageBus.GetProxy("two");

            Guid a, b, c, d;
            a = Guid.NewGuid();
            b = Guid.NewGuid();
            c = Guid.NewGuid();
            d = Guid.NewGuid();

            mbp1.Publish(new HashSet<Guid>() { a, b });
            mbp2.Subscribe(new HashSet<Guid>() { c, d });
            mbp1.Publish(new HashSet<Guid>() { c, d });
            mbp1.Send(new Measurement(c, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                    new StringValue("stop")));
            IMeasurement[] res = mbp2.Recv();
            Assert.AreEqual("stop", res[0].Value.GetRaw());
        }
        [TestMethod]
        public void SubBeforePubWithNoProducerExisting()
        {
            IMessageBusProxy mbp1 = messageBus.GetProxy("one");
            IMessageBusProxy mbp2 = messageBus.GetProxy("two");

            Guid a, b, c, d;
            a = Guid.NewGuid();
            b = Guid.NewGuid();
            c = Guid.NewGuid();
            d = Guid.NewGuid();

            mbp2.Subscribe(new HashSet<Guid>() { c, d });
            mbp1.Publish(new HashSet<Guid>() { c, d });
            mbp1.Send(new Measurement(c, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                    new StringValue("stop")));
            IMeasurement[] res = mbp2.Recv();
            Assert.AreEqual("stop", res[0].Value.GetRaw());
        }
        [TestMethod]
        public void TestThreads()
        {
            IMessageBusProxy mbp = messageBus.GetProxy("foo");
            IMessageBusProxy mbp2 = messageBus.GetProxy("bar");
            int rounds = 1000000;
            Guid g1 = Guid.NewGuid();
            HashSet<Guid> pubs = new HashSet<Guid>() { g1, Guid.NewGuid(), Guid.NewGuid() };
            mbp.Publish(pubs);
            mbp2.Subscribe(new HashSet<Guid>() { g1 });
            Task prod = new Task(() =>
            {
                Random rng = new Random();
                for (int i = 0; i < rounds; i++)
                {
                    IMeasurement m = new Measurement(g1, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                      new StringValue(rng.Next().ToString()));
                    mbp.Send(m);
                }
                mbp.Send(new Measurement(g1, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                    new StringValue("stop")));
            });

            Task<int> cons = new Task<int>(() =>
            {
                bool run = true;
                int count = 0;
                while (run)
                {
                    IMeasurement[] recv = mbp2.Recv();
                    count += recv.Length;
                    foreach (IMeasurement M in recv)
                    {
                        if (M.Value.GetRaw() == "stop") run = false;
                    }

                }
                return count;
            });
            cons.Start();
            Thread.Sleep(30);
            prod.Start();
            cons.Wait();
            prod.Wait();
            Assert.AreEqual(rounds + 1, cons.Result);
        }

        [TestMethod]
        public void TestUnsubscribe()
        {
            IMessageBusProxy mbp = messageBus.GetProxy("foo");
            IMessageBusProxy mbp2 = messageBus.GetProxy("bar");
            int rounds = 1000000;
            bool runconsumer = true;
            Guid g1 = Guid.NewGuid();
            HashSet<Guid> pubs = new HashSet<Guid>() { g1, Guid.NewGuid(), Guid.NewGuid() };
            mbp.Publish(pubs);
            mbp2.Subscribe(new HashSet<Guid>() { g1 });
            Task<int> prod = new Task<int>(() =>
            {
                Random rng = new Random();
                int sent = 0;
                for (int i = 0; i < rounds; i++)
                {
                    IMeasurement m = new Measurement(g1, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                      new StringValue(rng.Next().ToString()));
                    mbp.Send(m);
                    sent++;
                }
                runconsumer = false;
                return sent;
            });

            Task<int> cons = new Task<int>(() =>
            {

                int count = 0;
                bool subbed = true;
                while (runconsumer)
                {
                    IMeasurement[] recv = mbp2.Recv();
                    count += recv.Length;
                    if (subbed && count >= rounds / 2)
                    {
                        mbp2.UnSubscribe(new HashSet<Guid>() { g1 });
                        subbed = false;
                    }
                    foreach (IMeasurement M in recv)
                    {
                        if (M.Value.GetRaw() == "stop") runconsumer = false;
                    }

                }
                return count;
            });
            cons.Start();
            prod.Start();
            cons.Wait();
            prod.Wait();
            Assert.AreNotEqual(prod.Result, cons.Result);
            double tot_sent = (double)prod.Result;
            double tot_recv = (double)cons.Result;
            double ratio = tot_sent / tot_recv;
            Assert.AreEqual(ratio, 2.0, 0.1);
        }
        [TestMethod]
        public void TestChain()
        {
            int nToSend = 1000000;
            bool sourceActive = true;
            bool middleActive = true;
            IMessageBusProxy mbp1 = messageBus.GetProxy("one");
            IMessageBusProxy mbp2 = messageBus.GetProxy("two");
            IMessageBusProxy mbp3 = messageBus.GetProxy("three");

            Guid mbp1_g1 = Guid.NewGuid();
            Guid mbp2_g1 = Guid.NewGuid();
            Guid mbp2_g2 = Guid.NewGuid();

            mbp1.Publish(new HashSet<Guid>() { mbp1_g1 });
            mbp2.Publish(new HashSet<Guid>() { mbp2_g1, mbp2_g2 });
            mbp3.Subscribe(new HashSet<Guid>() { mbp2_g1, mbp2_g2 });
            mbp2.Subscribe(new HashSet<Guid>() { mbp1_g1 });
            bool onepub = false;


            // first proxy sends measurements in the form of "N,randfloat"
            Task t1 = new Task(() =>
            {
                Random rng = new Random();
                //Trace.WriteLine("starting mbp1");
                for (int i = 1; i <= nToSend; i++)
                {
                    IMeasurement m = new Measurement(mbp1_g1, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                        new StringValue(String.Format("{0},{1}", i.ToString(), rng.Next().ToString())));
                    //Trace.WriteLine("mpb1 sending");
                    mbp1.Send(m);
                    //if (i > 10) throw new Exception("got here!");
                    onepub = true;
                }
                sourceActive = false;
            });

            // second proxy splits into 2 ints, n, rn
            Task t2 = new Task(() =>
            {
                bool still_receiving = true;
                char[] delim = { ',' };
                while (still_receiving)
                {
                    int n, rn;
                    IMeasurement[] incoming = mbp2.Recv();

                    if (incoming.Length == 0)
                    {
                        if (!sourceActive)
                        {
                            still_receiving = false;
                            break;
                        }
                    }
                    //throw new Exception(string.Format("got here! {0}", incoming.Length));
                    foreach (IMeasurement m in incoming)
                    {
                        string v = m.Value.GetRaw();
                        string[] pair = v.Split(delim);
                        mbp2.Send(new Measurement(mbp2_g1, m.Timestamp, m.StateFlags,
                            new IntegerValue(int.Parse(pair[0]))));
                        mbp2.Send(new Measurement(mbp2_g2, m.Timestamp, m.StateFlags,
                            new IntegerValue(int.Parse(pair[1]))));
                    }

                }
                middleActive = false;
            });

            Task<double> t3 = new Task<double>(() =>
            {
                bool still_receiving = true;
                int rn_recv = 0;
                int n_recv = 0;
                int total_recv = 0;

                while (still_receiving)
                {
                    IMeasurement[] incoming = mbp3.Recv();
                    if (incoming.Length == 0)
                    {
                        if (!middleActive)
                        {
                            still_receiving = false;
                            break;
                        }
                    }
                    total_recv += incoming.Length;
                    foreach (IMeasurement m in incoming)
                    {
                        if (m.StreamID == mbp2_g1)
                        {
                            if ((n_recv + 1 != m.Value.GetRaw()))
                            {
                                throw new IndexOutOfRangeException(string.Format("Out of order! expected {0} got {1}",
                                    n_recv + 1, m.Value.GetRaw()));
                            }
                            n_recv = m.Value.GetRaw();
                        }
                        else if (m.StreamID == mbp2_g2)
                        {
                            rn_recv += 1;
                        }
                        else
                        {
                            throw new IndexOutOfRangeException("WTF - I shouldn't have gotten this!");
                        }
                    }
                }
                return ((double)total_recv) / ((double)rn_recv);
            });

            t2.Start();
            t3.Start();
            t1.Start();
            Task.WaitAll(t1, t2, t3);
            Assert.AreEqual(t3.Result, 2.0, .1);
        }

        [TestMethod]
        public void ManyReadersAndSendMany()
        {
            int nreaders = 4;
            int nToSend = 1000000;
            bool feeding = true;
            IMessageBusProxy feeder = messageBus.GetProxy("feeder");
            Guid val1 = Guid.NewGuid();
            Guid val2 = Guid.NewGuid();
            Guid val3 = Guid.NewGuid();
            Guid[] allVals = { val1, val2, val3 };
            feeder.Publish(new HashSet<Guid>(allVals));
            IMessageBusProxy[] readers = new IMessageBusProxy[nreaders];
            for (int i = 0; i < nreaders; i++)
            {
                readers[i] = messageBus.GetProxy(string.Format("reader{0}", i));

                readers[i].Subscribe(new HashSet<Guid>(allVals));
                //readers[i].Subscribe(new HashSet<Guid>() { val1, allVals[(i % 2)+1] });
            }

            Task<int> feedTask = new Task<int>(() =>
            {
                int nsent = 0;
                Random rng = new Random();
                for (int i = 0; i < nToSend; i++)
                {
                    IMeasurement[] sending = new IMeasurement[allVals.Length];
                    for (int j = 0; j < allVals.Length; j++)
                    {
                        sending[j] = new Measurement(allVals[j], new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                            new IntegerValue(rng.Next()));

                    }
                    feeder.Send(sending);
                    nsent += allVals.Length;
                }
                feeding = false;
                return nsent;
            });

            Task<int>[] allTasks = new Task<int>[nreaders + 1];
            allTasks[0] = feedTask;
            for (int i = 0; i < nreaders; i++)
            {
                int tmp = i;
                allTasks[i + 1] = new Task<int>(() =>
                {
                    IMessageBusProxy myproxy = (IMessageBusProxy)readers[tmp];
                    int nrecv = 0;
                    while (true)
                    {
                        IMeasurement[] res = myproxy.Recv();
                        if (res.Length <= 0)
                        {
                            if (!feeding)
                            {
                                break;
                            }
                        }
                        nrecv += res.Length;
                        foreach (IMeasurement m in res)
                        {
                            int x = m.Value.GetRaw() * 22;
                        }
                    }
                    return nrecv;
                });

            }
            foreach (Task t in allTasks)
            {
                t.Start();
            }
            Task.WaitAll(allTasks);
            Assert.AreEqual(nToSend * allVals.Length, allTasks[0].Result, "0");
            for (int i = 1; i <= nreaders; i++)
            {
                Assert.AreEqual(nToSend * allVals.Length, (allTasks[i]).Result, i.ToString());
            }
        }
        [TestMethod]
        public void LateSubscribe()
        {
            IMessageBusProxy sender = messageBus.GetProxy("sender");
            IMessageBusProxy recver = messageBus.GetProxy("recver");
            HashSet<Guid> allSignals = new HashSet<Guid>() { Guid.NewGuid(), Guid.NewGuid() };
            sender.Publish(allSignals);
            Random rng = new Random();
            for (int i = 0; i < 1000; i += 2)
            {
                IMeasurement[] send = new IMeasurement[allSignals.Count];
                int j = 0;
                foreach (Guid signal in allSignals)
                {
                    send[j] = new Measurement(signal, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                        new IntegerValue(rng.Next()));
                    j++;
                }
                sender.Send(send);
            }
            recver.Subscribe(allSignals);
            IMeasurement[] recv = recver.Recv();
            Assert.AreEqual(0, recv.Length);
            {
                IMeasurement[] send = new IMeasurement[allSignals.Count];
                int j = 0;
                foreach (Guid signal in allSignals)
                {
                    send[j] = new Measurement(signal, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                        new IntegerValue(rng.Next()));
                    j++;
                }
                sender.Send(send);
            }
            recv = recver.Recv();
            Assert.AreEqual(2, recv.Length);
        }
        [TestMethod]
        public void ThreadedResubscription()
        {
            // thread control
            bool run = true;
            bool resubOn = false;
            int resubCycles = 20;
            bool startedSending = false;
            Guid theSignal = Guid.NewGuid();
            //threads
            //thread 1 feed measurements
            Task<int> feedTask = new Task<int>(() =>
            {
                IMessageBusProxy feeder = messageBus.GetProxy("feeder");
                feeder.Publish(new HashSet<Guid>() { theSignal });
                int mcount = 1;
                while (run)
                {
                    startedSending = true;
                    IMeasurement m = new Measurement(theSignal, new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                        new IntegerValue(mcount));
                    feeder.Send(m);
                    mcount++;
                }
                return mcount;
            });
            //thread 2 constant consumption - checks that resub doesnt cause data loss
            Task<int> steadyTask = new Task<int>(() =>
            {
                IMessageBusProxy steady = messageBus.GetProxy("steady");
                bool subbed = false;
                int nRecv = 0;
                while (true)
                {
                    if (!startedSending) continue;
                    else
                    {
                        if (!subbed)
                        {
                            steady.Subscribe(new HashSet<Guid>() { theSignal });
                            subbed = true;
                        }
                    }
                    IMeasurement[] m = steady.Recv();
                    if (!run && m.Length == 0)
                    {
                        break;
                    }
                    foreach (IMeasurement i in m)
                    {
                        if (nRecv == 0)
                        {
                            nRecv = i.Value.GetRaw();
                        }
                        nRecv++;
                    }
                }
                return nRecv;
            });
            // thread 3 resubs, subscribes on and off pwm style, should get approx 1/2 of points sent
            Task<int> resubsTask = new Task<int>(() =>
            {
                IMessageBusProxy resubs = messageBus.GetProxy("resubs");
                bool subbed = false;
                int mRecv = 0;
                HashSet<Guid> subSet = new HashSet<Guid>() { theSignal };
                while (run)
                {
                    if (resubOn)
                    {
                        if (!subbed)
                        {
                            resubs.Subscribe(subSet);
                            subbed = true;
                        }
                        IMeasurement[] recv = resubs.Recv();
                        foreach (IMeasurement m in recv)
                        {
                            mRecv++;
                        }
                    }
                    else
                    {
                        if (subbed)
                        {
                            resubs.UnSubscribe(subSet);
                            subbed = false;
                        }
                        Thread.Yield();
                    }
                }
                return mRecv;
            });
            // Timer stuff
            System.Timers.Timer subPWM = new System.Timers.Timer(50);
            subPWM.Elapsed += ((source, args) =>
            {
                if (resubCycles > 0)
                {
                    if (startedSending)
                    {
                        resubOn = !resubOn;
                        resubCycles--;
                    }
                }
                else
                {
                    subPWM.Enabled = false;
                    run = false;
                }
            });
            subPWM.AutoReset = true;
            subPWM.Enabled = true;
            feedTask.Start();
            steadyTask.Start();
            resubsTask.Start();
            Task.WaitAll(feedTask, steadyTask, resubsTask);
            Assert.AreEqual(feedTask.Result, steadyTask.Result);
            // since we have a 50% duty cycle for the resub, we should have about half the sent
            // messages in our count. generous leeway because we aren't testing timers, just resubscription
            Assert.AreEqual(0.5, ((float)resubsTask.Result) / ((float)feedTask.Result), 0.1);
        }
        [TestMethod]
        public void UnsubThingsThatAreAvailableButNotCurrentlySubscribed()
        {
            IMessageBusProxy mbp1 = messageBus.GetProxy("one");
            IMessageBusProxy mbp2 = messageBus.GetProxy("two");

            Guid g1, g2;
            g1 = Guid.NewGuid();
            g2 = Guid.NewGuid();

            ISet<Guid> cycler = new HashSet<Guid>(){g1};
            ISet<Guid> allguids = new HashSet<Guid>(){g1,g2};
            
            // publish then sub then unsup n times
            mbp1.Publish(new HashSet<Guid>(){g2, g1});
            mbp2.Subscribe(cycler);
            mbp2.UnSubscribe(cycler);
            mbp2.UnSubscribe(cycler);

            // sub then usub it and one that was never subbed
            mbp2.Subscribe(cycler);
            mbp2.UnSubscribe(allguids);

            // other cases?
        }
        [TestMethod]
        public void TestBigPubSub()
        {
            IMessageBusProxy mbp1 = messageBus.GetProxy("one");
            IMessageBusProxy mbp2 = messageBus.GetProxy("two");
            IMessageBusProxy mbp3 = messageBus.GetProxy("three");
            IMessageBusProxy mbp4 = messageBus.GetProxy("four");
            IMessageBusProxy mbp5 = messageBus.GetProxy("five");

            mbp5.Publish(new HashSet<Guid>(makeGuids(15)));
            Guid[] pair1 = makeGuids(1);
            mbp4.Subscribe(new HashSet<Guid>(pair1));
            mbp3.Publish(new HashSet<Guid>(pair1));

            Guid[] bigSet = makeGuids(15);
            mbp1.Subscribe(new HashSet<Guid>(bigSet));
            mbp2.Publish(new HashSet<Guid>(bigSet));
            mbp2.Send(new Measurement(bigSet[4], new GSF.Ticks(), new SIEGate.Core.Legacy.MeasurementStateFlags(),
                   new StringValue("stop")));
            IMeasurement[] res = mbp1.Recv();
            Assert.AreEqual("stop", res[0].Value.GetRaw());
           
        }


        // helper methods
        private Guid[] makeGuids(int n)
        {
            Guid[] res = new Guid[n];
            for (int i=0; i < n; i++)
            {
                res[i] = Guid.NewGuid();
            }
            return res;
        }
    }
}
